/*
 * Copyright Hundsun Technologies Inc. All Rights Reserved.
 */

package selector

import (
	"context"
	"encoding/json"

	"hundsun.com/hsl/hschain/common/util/pb"
	"hundsun.com/hsl/hschain/protos"

	"go.uber.org/atomic"
	"hundsun.com/hsl/hschain/common/config"
	"hundsun.com/hsl/hschain/common/log/logging"
	"hundsun.com/hsl/hschain/common/message"
	concomm "hundsun.com/hsl/hschain/consensus/common"
	dposcomm "hundsun.com/hsl/hschain/consensus/impl/dpos/common"
	"hundsun.com/hsl/hschain/protos/ledger"
	dbm "hundsun.com/hsl/hschain/store/db"
)

const (
	goroutineTagVoteCounter                    = "vote-counter"
	goroutineTagVoter                          = "voter"
	configParamCompeteMiner                    = "competeMiner"
	configParamStartRound                      = "startRound"
	configParamEndRound                        = "endRound"
	configParamVoterAddress                    = "voterAddress"
	configParamVoterPeerID                     = "voterPeerID"
	configParamAgentAddress                    = "agentAddress"
	configParamVoteCounterSleepMillis          = "voteCounterSleepMillis"
	configParamVoteCounterBlocksFetchBatchSize = "voteCounterBlocksFetchBatchSize"
)

// NewSelector 新建一个selector
func NewSelector(client *concomm.BaseClient) *DposSelector {
	m := &DposSelector{
		client: client,
	}

	m.client.SubExecutedBlockFunc(func(block *ledger.ExecutedBlock) error {
		mqLog := m.log.WithField(logging.LogFieldMessageTopic, message.TopicExecutedBlock)
		mqLog.Infof("subscribing")
		blkLog := mqLog.WithField(logging.LogFieldMessageTopic, message.TopicExecutedBlock).
			WithField(logging.LogFieldBlockNumber, block.BlockExecResult.BlockNumber)
		roundOfBlock, _, _ := dposcomm.CalculateRound(block.Block.Header.BlockTime)
		if !m.voteCounter.maxExecBlock.Exists() || roundOfBlock > m.voteCounter.maxExecBlock.GetRound() {
			m.voteCounter.maxExecBlock.Store(true, roundOfBlock, block.BlockExecResult.BlockNumber)
			blkLog.Debugf("maxRoundOfExecutedBlocks set to '%d'", roundOfBlock)
		} else {
			blkLog.Debugf("round of message is the same as maxRoundOfExecutedBlocks: %d", roundOfBlock)
		}
		return nil
	})

	return m
}

// DposSelector 出块节点选举模块
type DposSelector struct {
	electionProgress *AtomicElectionContext
	voter            *Voter
	voteCounter      *VoteCounter

	config  *config.ModuleConfig
	client  *concomm.BaseClient
	rootCtx context.Context
	log     logging.Log
}

// Start 启动Selector
func (m *DposSelector) Start(ctx context.Context, cfg *config.ModuleConfig, log logging.Log) {
	m.config = cfg
	m.log = log
	m.rootCtx, _ = context.WithCancel(ctx)

	err := m.syncElectionContextFromMQ()
	if err != nil {
		m.log.Errorf("sync election context from MQ failed, err: %s", err.Error())
		return
	}

	m.voteCounter = &VoteCounter{
		log:    m.log.WithField(logging.LogFieldGoroutine, goroutineTagVoteCounter),
		module: m,
	}
	err = m.voteCounter.Init()
	if err != nil {
		m.log.Errorf("init vote counter failed, err: %s", err.Error())
		return
	}

	m.CountVotes()

	voterCtx, _ := context.WithCancel(m.rootCtx)
	m.voter = &Voter{
		log:    m.log.WithField(logging.LogFieldGoroutine, goroutineTagVoter),
		module: m,
		ctx:    voterCtx,
	}

	//启动投票goroutine
	m.Vote()
}

func (m *DposSelector) syncElectionContextFromMQ() error {
	var electionContext = &ElectionContext{}
	replyElectionBytes, err := m.client.RequestElectionContextInfo()
	if len(replyElectionBytes) > 0 {
		err = json.Unmarshal(replyElectionBytes, electionContext)
		if err != nil {
			m.log.Errorf("marshal election context info failed, err:%s\n", err.Error())
			return err
		}
	} else {
		electionContext = new(ElectionContext)
		electionContext.Round, _, _ = dposcomm.CalculateRound(pb.GetGenesisBlock().Header.BlockTime)
		electionContext.FirstBlockOfRound = 0
		electionContext.Candidates = make(map[protos.Address]*Candidate)
		electionContext.Delegations = make(map[protos.Address]*Delegation)
	}

	m.log.Infof("ElectionContext synchronized round:%d firstBlockOfRound:%d", electionContext.Round, electionContext.FirstBlockOfRound)
	electionContext.Log = m.log
	electionContext.LastHandledBlock = electionContext.FirstBlockOfRound
	m.electionProgress = NewAtomicElectionContext(electionContext)
	return nil
}

// GetLedgerDB 获取ledgerDB
func (m *DposSelector) GetLedgerDB() dbm.Database {
	return m.client.GetLedgerDB()
}

// GetStateDB 获取StateDB
func (m *DposSelector) GetStateDB() dbm.Database {
	return m.client.GetStateDB()
}

// CountVotes 计算选票
func (m *DposSelector) CountVotes() {
	voteCounterCtx, _ := context.WithCancel(m.rootCtx)
	go m.voteCounter.Run(voteCounterCtx)
}

// Vote 投票
func (m *DposSelector) Vote() {
	go m.voter.Run()
}

// NewAtomicElectionContext 新建AtomicElectionContext
func NewAtomicElectionContext(value *ElectionContext) *AtomicElectionContext {
	a := &AtomicElectionContext{}
	a.value.Store(value)
	return a
}

// AtomicElectionContext 结构定义
type AtomicElectionContext struct {
	value atomic.Value
}

// Load 加载ElectionContext
func (a *AtomicElectionContext) Load() *ElectionContext {
	return a.value.Load().(*ElectionContext)
}

// Store 存储ElectionContext
func (a *AtomicElectionContext) Store(value *ElectionContext) {
	a.value.Store(value)
}
